import threading
from kafka import KafkaConsumer, KafkaProducer
import time
import json
import queue


# kafka ip端口 修改为你使用的kafka server配置
KAFKA_HOST = "10.102.1.113:9092"
# 需消费数据topic
KAKOU_DATA_TOPIC = "kako_data"

pre_data_queue = queue.Queue(maxsize=2000)


# 接收实时数据
def receive_message_of_pre_data(topic):
    print("接收卡口车辆的实时数据", 'kafka 消费被触发')
    global pre_data_queue
    # 首先定义全局变量，用于在不同方法内互相通信
    consumer = KafkaConsumer(
        topic,
        bootstrap_servers=KAFKA_HOST,
        auto_offset_reset='latest',
        enable_auto_commit=False
    )

    # 将消费者的偏移量设置为最后一条消息的偏移量
    while True:
        time.sleep(0.1 / 20)
        for message in consumer:
            if message:
                try:
                    message_value = message.value.decode('utf-8')
                    # 根据发送数据不同的数据格式选择
                    vehicle_info = json.loads(message_value)
                    # TODO 接收到的数据，这里添加处理逻辑
                    print("接收数据", vehicle_info)
                    if pre_data_queue.qsize() < 2000:
                        pre_data_queue.put(vehicle_info)
                except Exception as e:
                    print(str(e))


threading.Thread(target=receive_message_of_pre_data, args=(KAKOU_DATA_TOPIC,)).start()
# threading.Thread(target=receive_message_of_pre_data,args=(KAKOU_DATA_TOPIC)).start()

json_path = r"D:\GJ\项目\a.json"
file_w = open(json_path, "w", encoding="utf-8")
while True:

    if pre_data_queue.qsize() > 0:
        info = pre_data_queue.get()
        file_w.write(json.dumps(info) + "\n")

    time.sleep(0.1 / 10)

# # kafka发送方法
# # 需要发动到的topic
# PRE_DATA_TOPIC = "PRE_DATA_TOPIC"
# # 发送数据
# now_frame_data = {}
# producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
#                          value_serializer=lambda x: json.dumps(x).encode('utf-8'))
# count = 0
# while True:
#     # 事件检测结果
#     now_frame_data = {"test":str(count)}
#     producer.send(PRE_DATA_TOPIC, value=now_frame_data, key="key")
#     print("发送数据",now_frame_data)
#     count += 1
#     time.sleep(0.1 / 10)
